#include <cassert>
#include "../include/LogSvr.h"
#include "../include/NDK.h"
#include "../include/MessageQueue.h"
#include "../include/ThreadManager.h"

MessageQueue mq;
typedef Singleton_T<LogSvr> log;
ThreadMutex log_lock;
void *prc1 (void *arg)
{
    int i = 0;
    int *p = 0;
    MessageBlock * mb = NULL;
    while (i++ < 100000)
    {
	mb = new MessageBlock ((char *)new int(i), sizeof (int));
	int ret = mq.enqueue_tail (mb);
	log::instance ()->put (LOG_RINFO, "put an data : %d, dataq size = %d\n", 
		i, mq.size ());
    }
    mb = new MessageBlock ((char *)new int(i), sizeof (int));
    mb->data_type (MessageBlock::MB_STOP);
    mq.enqueue_tail (mb);
    log::instance ()->put (LOG_RINFO, "put an data : %d, dataq size = %d\n", 
	    i, mq.size ());
    return 0;
}
void *prc2 (void *arg)
{
    MessageBlock * mb = NULL;
    int i = 0;
    while(1)
    {
	mq.dequeue_head (mb);
	if (mb)
	{
	    if (mb->data_type () == MessageBlock::MB_STOP)
	    {
		log::instance ()->put (LOG_RINFO, 
			"gut an data : %d and stop, dataq size = %d\n", 
			*reinterpret_cast<int*>(mb->base ()), mq.size ());
		break;
	    }
	}
	log::instance ()->put (LOG_RINFO, "gut an data : %d, dataq size = %d\n", 
		*reinterpret_cast<int*>(mb->base ()), mq.size ());
    }
    return 0;
}
void *prc3 (void *arg)
{
    int i = 0;
    int *p = 0;
    MessageBlock * mb = NULL;
    while (i++ < 100000)
    {
	mb = new MessageBlock ((char *)new int(i), sizeof (int));
	int ret = mq.enqueue_tail (mb);
	log::instance ()->put (LOG_RINFO, "put an data : %d, dataq size = %d\n", 
		i, mq.size ());
    }
    mb = new MessageBlock ((char *)new int(i), sizeof (int));
    mb->data_type (MessageBlock::MB_STOP);
    mq.enqueue_tail (mb);
    log::instance ()->put (LOG_RINFO, "put an data : %d, dataq size = %d\n", 
	    i, mq.size ());
    return 0;
}
#if 0
void *prc3 (void *arg)
{
    int i = 0;
    int *p = 0;
    MessageBlock * mb = NULL;
    MessageBlock * tmb = NULL;
    MessageBlock * head = NULL;
    while (i < 100000)
    {
	mb = new MessageBlock ((char *)new int(i), sizeof (int));
	head = mb;
	for (int n = 0; n < 10; ++n)
	{
	    i++;
	    tmb = new MessageBlock ((char *)new int(i), sizeof (int));
	    mb->next (tmb);	
	    mb = mb->next ();
	}
	int ret = mq.enqueue_tail (head);
	log::instance ()->put (LOG_RINFO, "put an data : %d, dataq size = %d\n", i, mq.size ());
    }
    mb = new MessageBlock ((char *)new int(i), sizeof (int));
    mb->data_type (MessageBlock::MB_STOP);
    mq.enqueue_tail (mb);
    log::instance ()->put (LOG_RINFO, "put an data : %d, dataq size = %d\n", i, mq.size ());
    return 0;
}
#endif
void *prc4 (void *arg)
{
    MessageBlock * mb = NULL;
    int i = 0;
    int count = 0;
    while(1)
    {
	int num = mq.dequeue_head_n (mb);
	if (num > 0)
	{
	    count += num;
	    for (MessageBlock *tmb = mb; tmb != 0; tmb = tmb->next ())
	    {
		if (tmb)
		{
		    if (tmb->data_type () == MessageBlock::MB_STOP)
		    {
			log::instance ()->put (LOG_RINFO, 
				"gut %d data : total %d and stop, "
				"dataq size = %d\n", num, count, mq.size ());
			return 0;
		    }
		    //delete mb->base ();
		}
	    }
	    log::instance ()->put (LOG_RINFO, "gut an data : %d, dataq size = %d\n", 
		    num, mq.size ());
	}
    }
    return 0;
}
void *prc5 (void *arg)
{
    int i = 0;
    int *p = 0;
    MessageBlock * mb = NULL;
    TimeValue tv(0, 5000);
    while (1)
    {
	for (int n = 0; n < 10; ++n)
	{
	    mb = new MessageBlock ((char *)new int(i), sizeof (int));
	    mq.enqueue_tail (mb);
	    //log::instance ()->put (LOG_RINFO, "put an data : %d, dataq size = %d\n", i, mq.size ());
	} 
	NDK::handle_ready (0, 0, 0, 0, &tv);
    }
    mb = new MessageBlock ((char *)new int(i), sizeof (int));
    mb->data_type (MessageBlock::MB_STOP);
    mq.enqueue_tail (mb);
    log::instance ()->put (LOG_RINFO, "put an data : %d, dataq size = %d\n", i, mq.size ());
    return 0;
}
void *prc6 (void *arg)
{
    MessageBlock * mb = NULL;
    int i = 0;
    int count = 0;
    while(1)
    {
	//int num = mq.dequeue_tail_n (mb, 5, -1);
	struct timeval  tp_before;
	struct timeval  tp_after;
	::gettimeofday (&tp_before, NULL);
	int num = mq.dequeue_head_n (mb, 1, 0);
	if (num > 0)
	{
	    count += num;
	    for (MessageBlock *tmb = mb; tmb != 0; tmb = tmb->next ())
	    {
		if (tmb->data_type () == MessageBlock::MB_STOP)
		{
		    log::instance ()->put (LOG_RINFO, "gut %d data : total %d and "
			    "stop, dataq size = %d\n", num, count, mq.size ());
		    //assert (count == 5);
		    return 0;
		}
		log::instance ()->put (LOG_RINFO, 
			"gut an data : %d, num = %d, dataq size = %d\n", 
			*reinterpret_cast<int*>(tmb->base ()), num, mq.size ());
		delete tmb->base ();
#if 0
		MessageBlock *tmp = tmb;
		tmb = tmb->next ();
		tmp->release ();
#endif
	    }
	    mb->release ();
	    mb = 0;
	}
	::gettimeofday (&tp_after, NULL);
	if (tp_before.tv_sec - tp_after.tv_sec > 1)
	{
	    if (tp_before.tv_usec - tp_after.tv_usec > 1000 *100)
	    {
		log::instance ()->put (LOG_RINFO, "timeout diffsec = %d, diffmsec = %d\n", 
			tp_before.tv_sec - tp_after.tv_sec,
			(tp_before.tv_usec - tp_after.tv_usec)*1000);
	    }
	}
    }
    return 0;
}
int main ()
{
    log::instance ()->open ("log"); 
    // Test destructor of MessageQueue
    MessageQueue *mq = new MessageQueue;
    for (int i = 0; i < 100; ++i)
    {
	MessageBlock *mb = new MessageBlock (sizeof (int));
	mq->enqueue_tail (mb);
    }
    delete mq;
    // end
#if 0
    ThreadManager::instance ()->spawn (prc1);
    ThreadManager::instance ()->spawn (prc2);
#endif
#if 0
    ThreadManager::instance ()->spawn (prc3);
    ThreadManager::instance ()->spawn (prc4);
#endif
    ThreadManager::instance ()->spawn (prc5);
    ThreadManager::instance ()->spawn (prc6);
    ThreadManager::instance ()->wait ();
}
